package com.yanfan.iot.service.impl;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson2.JSON;
import com.dtflys.forest.http.ForestRequest;
import com.yanfan.common.utils.DateUtils;
import com.yanfan.framework.manager.AsyncManager;
import com.yanfan.http.service.HttpClientFactory;
import com.yanfan.http.service.SuccessCondition;
import com.yanfan.iot.domain.Bridge;
import com.yanfan.iot.domain.HttpClient;
import com.yanfan.iot.domain.MqttClient;
import com.yanfan.iot.domain.MultipleDataSource;
import com.yanfan.iot.mapper.BridgeMapper;
import com.yanfan.iot.model.MqttClientConfig;
import com.yanfan.iot.ruleEngine.DataSourceService;
import com.yanfan.iot.ruleEngine.MqttBridgeClientFactory;
import com.yanfan.iot.service.IBridgeService;
import com.yanfan.iot.service.IHttpClientService;
import com.yanfan.iot.service.IMqttBridgeClientService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;

import java.sql.SQLException;
import java.util.List;
import java.util.Objects;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 数据桥接Service业务层处理
 *
 * @author zhuangpeng.li
 * @date 2024-06-06
 */
@Service
@Slf4j
public class BridgeServiceImpl implements IBridgeService, ApplicationRunner {
    @Autowired
    private BridgeMapper bridgeMapper;

    @Autowired
    private IHttpClientService httpClientService;

    @Autowired
    private IMqttBridgeClientService mqttClientService;

    @Autowired
    private DataSourceService dataSourceService;

    @Value("${spring.profiles.active:}")
    private String activeEnv;


    /**
     * 查询数据桥接
     *
     * @param id 数据桥接主键
     * @return 数据桥接
     */
    @Override
    public Bridge selectBridgeById(Long id) {
        return bridgeMapper.selectBridgeById(id);
    }

    /**
     * 查询数据桥接列表
     *
     * @param bridge 数据桥接
     * @return 数据桥接
     */
    @Override
    public List<Bridge> selectBridgeList(Bridge bridge) {
        return bridgeMapper.selectBridgeList(bridge);
    }

    /**
     * 新增数据桥接
     *
     * @param bridge 数据桥接
     * @return 结果
     */
    @Override
    public int insertBridge(Bridge bridge) {
        bridge.setCreateTime(DateUtils.getNowDate());
        int i = bridgeMapper.insertBridge(bridge);
        if (bridge.getType() == 4) {
            MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
            if (config != null) {
                try {
                    MqttAsyncClient client = MqttBridgeClientFactory.instance(mqttClientService.buildMqttClientConfig(config, bridge.getId()));
                    if (bridge.getConfigJson() != null && bridge.getDirection() == 1) {
                        bridge.setStatus("1");
                        MqttBridgeClientFactory.addSubscribe(client, bridge.getId(), bridge.getRoute());
                    }
                } catch (MqttException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        return i;
    }

    /**
     * 修改数据桥接
     *
     * @param bridge 数据桥接
     * @return 结果
     */
    @Override
    public int updateBridge(Bridge bridge) {
        bridge.setUpdateTime(DateUtils.getNowDate());
        Bridge dbBridge = bridgeMapper.selectBridgeById(bridge.getId());
        if (bridge.getType() == 4) {
            if (bridge.getConfigJson() != null && bridge.getDirection() == 1) {
                MqttBridgeClientFactory.delSubscribe(dbBridge.getId(), dbBridge.getRoute());
                MqttBridgeClientFactory.remove(dbBridge.getId());
                bridge.setStatus("1");
                MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
                try {
                    if (bridge.getConfigJson() != null && bridge.getDirection() == 1 && Objects.equals(bridge.getEnable(), "1")) {
                        MqttAsyncClient client = MqttBridgeClientFactory.instance(mqttClientService.buildMqttClientConfig(config, bridge.getId()));
                        bridge.setStatus("1");
                        MqttBridgeClientFactory.addSubscribe(client, bridge.getId(), bridge.getRoute());
                    }
                } catch (MqttException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        return bridgeMapper.updateBridge(bridge);
    }

    @Override
    public int connect(Bridge bridge) {
        AtomicInteger result = new AtomicInteger();
        if (bridge.getType() == 3) {
            HttpClient config = JSON.parseObject(bridge.getConfigJson(), HttpClient.class);
            ForestRequest request = HttpClientFactory.instance(httpClientService.buildHttpClientConfig(config));
            request.successWhen(SuccessCondition.class).onSuccess(((data, req, res) -> {
                bridge.setStatus("1");
                result.set(1);
            })).onError((ex, req, res) -> {
                bridge.setStatus("0");
                result.set(0);
            }).execute();
        } else if (bridge.getType() == 4) {
            MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
            if (config != null) {
                MqttAsyncClient client = null;
                try {
                    MqttClientConfig clientConfig = mqttClientService.buildMqttClientConfig(config, bridge.getId());
                    client = MqttBridgeClientFactory.instance(clientConfig);
                    if (client != null && client.isConnected()) {
                        bridge.setStatus("1");
                        result.set(1);
                    } else {
                        bridge.setStatus("0");
                        result.set(0);
                    }
                } catch (MqttException e) {
                    log.error("=>mqtt客户端创建错误 error:{}", e.getMessage());
                } finally {
                    if (Objects.isNull(bridge.getId()) && Objects.nonNull(client) && client.isConnected()) {
                        try {
                            client.disconnect();
                            client.close();
                        } catch (MqttException e) {
                            log.error("=>临时mqtt客户端错误 error:{}", e.getMessage());
                        }
                    }
                }
            }
        } else if (bridge.getType() == 5) {
            MultipleDataSource config = JSON.parseObject(bridge.getConfigJson(), MultipleDataSource.class);
            DruidDataSource dataSource = dataSourceService.createDataSource(config);
            try {
                if (dataSource != null && dataSource.getConnection() != null) {
                    bridge.setStatus("1");
                    result.set(1);
                } else {
                    bridge.setStatus("0");
                    result.set(0);
                }
            } catch (SQLException e) {
                log.error("=>数据源连接错误");
            }
        }
        bridgeMapper.updateBridge(bridge);
        return result.get();
    }

    /**
     * 批量删除数据桥接
     *
     * @param ids 需要删除的数据桥接主键
     * @return 结果
     */
    @Override
    public int deleteBridgeByIds(Long[] ids) {
        int i = 0;
        for (Long id : ids) {
            i += deleteBridgeById(id);
        }
        return i;
    }

    /**
     * 删除数据桥接信息
     *
     * @param id 数据桥接主键
     * @return 结果
     */
    @Override
    public int deleteBridgeById(Long id) {
        Bridge dbBridge = bridgeMapper.selectBridgeById(id);
        if (dbBridge == null) {
            return 0;
        }

        if (dbBridge.getType() == 4) {
            MqttBridgeClientFactory.delSubscribe(dbBridge.getId(), dbBridge.getRoute());
            MqttBridgeClientFactory.remove(dbBridge.getId());
        }
        return bridgeMapper.deleteBridgeById(id);
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (!Objects.equals(activeEnv, "prod")) {
            return;
        }

        AsyncManager.me().execute(new TimerTask() {
            @Override
            public void run() {
                List<Bridge> list = bridgeMapper.selectBridgeList(new Bridge());
                for (Bridge bridge : list) {
                    if (bridge.getType() == 4 && Objects.equals(bridge.getEnable(), "1")) {
                        MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class);
                        if (config != null) {
                            try {
                                MqttAsyncClient client = MqttBridgeClientFactory.instance(mqttClientService.buildMqttClientConfig(config, bridge.getId()));
                                if (client != null && bridge.getDirection() == 1 && !Objects.equals(bridge.getRoute(), "")) {
                                    MqttBridgeClientFactory.addSubscribe(client, bridge.getId(), bridge.getRoute());
                                }
                            } catch (MqttException e) {
                                throw new RuntimeException(e);
                            }

                        }
                    }
                }
            }
        });
    }
}
